package org.study.module.hadoop;

import org.study.module.common.AbstractConfigBuilder;
import org.study.module.common.ModuleCommon;

import java.util.ArrayList;
import java.util.List;

/**
 * HDFS配置，构建类
 * hdfs-site.xml
 */
public class HDFSConfigBuilder extends AbstractConfigBuilder {


    /**
     * 副本数量： 数字型字符串
     *
     * @param replicationNum 副本数量：如：3
     */
    public HDFSConfigBuilder replicationNum(String replicationNum) {
        configs.add("    <!-- 副本数 -->");
        ModuleCommon.addProperty("dfs.replication", replicationNum, configs);
        return this;
    }

    /**
     * NameNode 元数据存储路径（如 file:///data/hadoop/name），支持逗号分隔多路径备份元数据
     *
     * @param nameNodeDir NameNode 存储的位置
     */
    public HDFSConfigBuilder nameNodeDir(String nameNodeDir) {
        configs.add("    <!-- NameNode 存储的位置 -->");
        ModuleCommon.addProperty("dfs.namenode.name.dir", nameNodeDir, configs);
        return this;
    }

    /**
     * DataNode 数据块存储路径（如 file:///data/hadoop/data），多磁盘时可配置多个路径提升吞吐
     *
     * @param dataNodeDir DataNode 存储的位置
     */
    public HDFSConfigBuilder dataNodeDir(String dataNodeDir) {
        configs.add("    <!-- DataNode 存储的位置 -->");
        ModuleCommon.addProperty("dfs.datanode.data.dir", dataNodeDir, configs);
        return this;
    }

    /**
     *  DataNode 数据传输端口,
     * @param address 如：0.0.0.0:50010
     */
    public HDFSConfigBuilder dataNodeAddress(String address) {
        configs.add("    <!-- DataNode 数据传输端口 -->");
        ModuleCommon.addProperty("dfs.datanode.address", address, configs);
        return this;
    }

    /**
     * DataNode Web UI端口
     * @param address 如：0.0.0.0:50075
     * @return
     */
    public HDFSConfigBuilder dataNodeHttpAddress(String address) {
        configs.add("    <!-- DataNode Web UI端口 -->");
        ModuleCommon.addProperty("dfs.datanode.http.address", address, configs);
        return this;
    }

    /**
     * HDFS 块大小（默认 128MB），影响 MapReduce 任务并行度；设为 256MB（268435456）可减少 NN 内存压力。
     *
     * @param blockSize 块大小（默认 128MB）
     */
    public HDFSConfigBuilder blockSize(String blockSize) {
        configs.add("    <!-- HDFS 块大小（默认 128MB） -->");
        ModuleCommon.addProperty("dfs.blocksize", blockSize, configs);
        return this;
    }
    /**
     * 高可用：启用自动故障转移（需配合 ZooKeeper）
     */
    public HDFSConfigBuilder automaticFailover() {
        return automaticFailover("true");
    }
    /**
     * 高可用：启用自动故障转移（需配合 ZooKeeper）
     *
     * @param enabled 是否启用： true, false
     */
    public HDFSConfigBuilder automaticFailover(String enabled) {
        configs.add("    <!-- 自动故障转移 -->");
        ModuleCommon.addProperty("dfs.ha.automatic-failover.enabled", enabled, configs);
        return this;
    }

    /**
     * 故障切换代理类（固定值：org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider）
     * @param nameNodeHost 数据节点名称
     */
    public HDFSConfigBuilder failoverProProvider(String nameNodeHost) {
        configs.add("    <!-- 故障切换代理类 -->");
        ModuleCommon.addProperty("dfs.client.failover.proxy.provider."+nameNodeHost, "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider", configs);
        return this;
    }

    /**
     * 高可用: 逻辑命名服务名称,用于标识 HA 集群
     *
     * @param clusterName 服务名称 如 (mycluster）
     * @return
     */
    public HDFSConfigBuilder nameServices(String clusterName) {
        configs.add("    <!-- 集群服务名 -->");
        ModuleCommon.addProperty("dfs.nameservices", clusterName, configs);
        return this;
    }

    /**
     * 高可用：NameNode节点列表,
     *
     * @param clusterName 集群名称，参考nameServices方法
     * @param nns         如 (nn1,nn2）
     * @return
     */
    public HDFSConfigBuilder haNameNodes(String clusterName, String nns) {
        configs.add("    <!-- NameNode节点列表 -->");
        ModuleCommon.addProperty("dfs.ha.namenodes." + clusterName, nns, configs);
        return this;
    }

    /**
     * 高可用：NameNode节点的rcp地址，多个的时候需要添加多次
     *
     * @param clusterName  集群名称
     * @param nameNodeHost NameNode节点地址：如（bigdatanameNode）
     * @param rpcPort      RPC服务的端口号： 如（8020）
     */
    public HDFSConfigBuilder nameNodeRpcAddress(String clusterName, String nameNodeHost, String rpcPort) {
        addMessage(nameNodeHost+",NameNode的RPC地址");
        addProperty("dfs.namenode.rpc-address." + clusterName + "." + nameNodeHost, nameNodeHost + ":" + rpcPort);
        return this;
    }

    /**
     * 高可用：NameNode节点的http地址，多个的时候需要添加多次
     *
     * @param clusterName  集群名称
     * @param nameNodeHost NameNode节点地址：如（bigdatanameNode）
     * @param httpPort     HTTP服务的端口号：如（9870）
     */
    public HDFSConfigBuilder nameNodeHttpAddress(String clusterName, String nameNodeHost, String httpPort) {
        addMessage(nameNodeHost+",NameNode的HTTP地址");
        addProperty("dfs.namenode.http-address." + clusterName + "." + nameNodeHost, nameNodeHost + ":" + httpPort);
        return this;
    }
    /**
     * 高可用：NameNode节点的https地址
     *
     * @param clusterName  集群名称
     * @param nameNodeHost NameNode节点地址：如（bigdatanameNode）
     * @param httpPort     HTTP服务的端口号：如（50470）
     */
    public HDFSConfigBuilder nameNodeHttpsAddress(String clusterName, String nameNodeHost, String httpPort) {
        addMessage(nameNodeHost+",NameNode的HTTPS地址");
        addProperty("dfs.namenode.https-address." + clusterName + "." + nameNodeHost, nameNodeHost + ":" + httpPort);
        return this;
    }

    /**
     * JournalNode 集群地址，（元数据同步）
     *
     * @param editsDir    集群地址 如（node1:8485;node2:8485;node3:8485）
     * @param clusterName 集群名称
     */
    public HDFSConfigBuilder nameNodeSharedEditsDir(String editsDir, String clusterName) {
        configs.add("    <!-- JournalNode 集群地址 -->");
        ModuleCommon.addProperty("dfs.namenode.shared.edits.dir", "qjournal://" + editsDir + "/" + clusterName, configs);
        return this;
    }

    /**
     * JournalNode 节点数据日志地址
     *
     * @param journalNodeDir 节点数据日志地址
     */
    public HDFSConfigBuilder journalNodeEditDir(String journalNodeDir) {
        configs.add("    <!-- JournalNode 节点数据日志地址 -->");
        ModuleCommon.addProperty("dfs.journalnode.edits.dir", journalNodeDir, configs);
        return this;
    }

    /**
     * 若启用 Kerberos，需添加以下配置（替换REALM为实际域名）
     *
     * @param realm 如：CMS.COM
     */
    public HDFSConfigBuilder nameNodeKerberosPrincipal(String installPath,String host,String realm) {
        configs.add("    <!-- NameNode 节点添加 kerberos认证 -->");
        ModuleCommon.addProperty("dfs.namenode.kerberos.principal", "nn/"+host+"@"+realm, configs);
        configs.add("    <!-- 对应节点的keytab路径 -->");
        ModuleCommon.addProperty("dfs.namenode.keytab.file", HadoopShell.getHadoopKerberos(installPath)+"/hdfs-"+host+".keytab", configs);
        return this;
    }
    public HDFSConfigBuilder blockAccessTokenEnable() {
        configs.add("    <!-- 启用块访问令牌 -->");
        ModuleCommon.addProperty("dfs.block.access.token.enable", "true", configs);
        return this;
    }
    /**
     * 若启用 Kerberos，需添加以下配置（替换REALM为实际域名）
     *
     * @param realm 如：CMS.COM
     */
    public HDFSConfigBuilder dataNodeKerberosPrincipal(String installPath,String host,String realm) {
        configs.add("    <!-- DataNode 节点添加 kerberos认证 -->");
        ModuleCommon.addProperty("dfs.datanode.kerberos.principal", "dn/"+host+"@"+realm, configs);
        configs.add("    <!-- 对应节点的keytab路径 -->");
        ModuleCommon.addProperty("dfs.datanode.keytab.file", HadoopShell.getHadoopKerberos(installPath)+"/hdfs-"+host+".keytab", configs);
        return this;
    }
    /**
     * 若启用 Kerberos，需添加以下配置（替换REALM为实际域名）
     *
     * @param realm 如：CMS.COM
     */
    public HDFSConfigBuilder journalNodeKerberosPrincipal(String installPath,String host,String realm) {
        configs.add("    <!-- JournalNode 节点添加 kerberos认证 -->");
        ModuleCommon.addProperty("dfs.journalnode.kerberos.principal", "jn/"+host+"@"+realm, configs);
        configs.add("    <!-- 对应节点的keytab路径 -->");
        ModuleCommon.addProperty("dfs.journalnode.keytab.file", HadoopShell.getHadoopKerberos(installPath)+"/hdfs-"+host+".keytab", configs);
        return this;
    }

    /**
     * 若启用 Kerberos，需添加以下配置（替换REALM为实际域名）
     *
     * @param realm 如：CMS.COM
     */
    public HDFSConfigBuilder zkfcKerberosPrincipal(String installPath,String host,String realm) {
        configs.add("    <!-- ZKFC 节点添加 kerberos认证 -->");
        ModuleCommon.addProperty("dfs.zkfc.kerberos.principal", "zkfc/"+host+"@"+realm, configs);
        configs.add("    <!-- 对应节点的keytab路径 -->");
        ModuleCommon.addProperty("dfs.zkfc.keytab.file", HadoopShell.getHadoopKerberos(installPath)+"/hdfs-"+host+".keytab", configs);
        return this;
    }

    public static List<String> hdfsSiteHA(String installPath, String nameNodeHost, String clusterName, String nns) {
        List<String> configs = new ArrayList<>();

        configs.add("    <!-- 隔离机制 -->");
        ModuleCommon.addProperty("dfs.ha.fencing.methods", "sshfence", configs);
        ModuleCommon.addProperty("dfs.ha.fencing.ssh.private-key-files", "/home/hadoop/.ssh/id_rsa", configs);

        return configs;
    }

    /**
     * KMS服务
     * @param hostAndPort kms服务地址和对应的端口号，如：node3:9600
     * @return
     */
    public HDFSConfigBuilder kms(String hostAndPort) {
        configs.add("    <!-- KMS服务地址 -->");
        ModuleCommon.addProperty("dfs.encryption.key.provider.uri", "kms://@"+hostAndPort+"/kms", configs);
        configs.add("    <!-- 启用加密区域 -->");
        ModuleCommon.addProperty("dfs.encryption.zones.enabled", "true", configs);
        configs.add("    <!-- 密钥缓存大小 -->");
        ModuleCommon.addProperty("dfs.encryption.key.provider.cache.size", "100", configs);
        return this;
    }


    public void addConfigHead() {
        configs.add("cat > $HADOOP_HOME/etc/hadoop/hdfs-site.xml << EOF");
        super.addConfigHead();
    }

}
